package com.h2cloud.ReactorTest.reactor.server.multi;

import com.h2cloud.ReactorTest.reactor.server.AsynHandler;
import com.h2cloud.ReactorTest.reactor.server.Handler;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Package: com.h2cloud.ReactorTest.reactor.server.multi
 * @ClassName: MainAcceptor
 * @Author: wulongkai
 * @CreateTime: 2024/7/23 16:04
 * @Description:
 */
public class MainAcceptor implements Runnable {

    // 指定从Reactor一共有16个
    private static final int TOTAL_SUBREACTOR_NUM = 16;

    // 服务端的listen-socket管道
    private final ServerSocketChannel listenSocketChannel;

    // 用于运行从Reactor
    private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            TOTAL_SUBREACTOR_NUM, TOTAL_SUBREACTOR_NUM * 2,
            60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));

    // 从Reactor集合
    private final List<SubReactor> subReactors = new ArrayList<>(TOTAL_SUBREACTOR_NUM);

    public MainAcceptor(ServerSocketChannel listenSocketChannel) throws IOException {
        this.listenSocketChannel = listenSocketChannel;
        // 将从Reactor初始化出来并运行
        for (int i = 0; i < TOTAL_SUBREACTOR_NUM; i++) {
            SubReactor subReactor = new SubReactor(Selector.open());
            subReactors.add(subReactor);
            threadPool.execute(subReactor);
        }
    }

    @Override
    public void run() {
        try {
            // 为连接的客户端创建client-socket管道
            SocketChannel clientSocketChannel = listenSocketChannel.accept();
            // 设置为非阻塞
            clientSocketChannel.configureBlocking(false);
            // 任意选择一个从Reactor，让其监听连接的客户端的READ事件
            Optional<SubReactor> anySubReactor = subReactors.stream().findAny();
            if (anySubReactor.isPresent()) {
                SubReactor subReactor = anySubReactor.get();
                // 从Reactor的多路复用器会阻塞在select()方法上
                // 这里需要先唤醒多路复用器，立即从select()方法返回
                subReactor.getSelector().wakeup();
                // 让从Reactor负责处理客户端的READ事件
                clientSocketChannel.register(subReactor.getSelector(), SelectionKey.OP_READ,
                        new AsynHandler(clientSocketChannel));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

